package cn.doitedu.day08

import java.text.SimpleDateFormat
import java.util.Calendar

import scala.io.Source

/**
 * 统计连续登录3天及以上的用户
 */
object Test02 {

  def main(args: Array[String]): Unit = {

    val source = Source.fromFile("data/test2.txt")

    val lines: Iterator[String] = source.getLines()
    //对数据进行整理
    val tuples: Iterator[(String, String)] = lines.map(line => {
      val fields = line.split(",")
      val uid = fields(0)
      val dt = fields(1)
      (uid, dt)
    })
    //将同一个用户的数据分到一个组内
    //迭代器是无法分组的，不能调用groupBy方法
    //迭代器无法groupBy，groupBy的本质是将数据保存到Map中，key就是分组的条件，values是条件相同的其他数据
    val grouped: Map[String, List[(String, String)]] = tuples.toList.distinct.groupBy(_._1) //按照用户Id进行分组
    //对同一个用户的数据按照日期进行排序
    val values: Iterable[List[(String, String)]] = grouped.values

    val sdf = new SimpleDateFormat("yyyy-MM-dd")
    val calendar = Calendar.getInstance()

    val uidDtAndDiff: Iterable[(String, String, String)] = values.flatMap(lst => {
      //按照日期排序（字符串）
      val sorted: Seq[(String, String)] = lst.sortBy(_._2)
      var i = 0
      //对排序的数据进行遍历
      sorted.map(t => {
        i += 1
        val uid = t._1
        val dt = t._2
        val date = sdf.parse(dt)
        //使用Calender
        calendar.setTime(date)
        //同当前的时间减去行号
        calendar.add(Calendar.DATE, -i)
        //返回相减后的差值
        val time = calendar.getTime
        //再将差值转成yyyy-MM-dd
        val diff: String = sdf.format(time)
        //用户ID，登录日期，差值
        (uid, dt, diff)
      })
    })
    //将数据再分组
    //(将用户ID相同，并且日期差值)相同的放入到同一个组中
    val grpued2: Map[(String, String), Iterable[(String, String, String)]] = uidDtAndDiff.groupBy(t => (t._1, t._3))

    val res = grpued2.values.map(it => {
      //按照日期排序
      val sorted = it.toList.sortBy(_._2)
      val uid = sorted.map(_._1).min
      //起始日期
      val startTime = sorted.map(_._2).min
      //结束日期
      val endTime = sorted.map(_._2).max
      //集合的长度就是天数
      val size = sorted.size
      (uid, startTime, endTime, size)
    }).filter(_._4 >= 3)

    println(res.toBuffer)


    source.close()

  }

}
